package com.example.mq2.mqserver;

import com.example.mq2.common.MqException;
import com.example.mq2.mqserver.core.*;
import com.example.mq2.mqserver.datacenter.DiskDataCenter;
import com.example.mq2.mqserver.datacenter.MemoryDataCenter;
import com.sun.org.apache.xerces.internal.impl.XMLEntityScanner;
import org.apache.ibatis.annotations.Param;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class VirtualHost {

    private String virtualHostName;
    private MemoryDataCenter memoryDataCenter = new MemoryDataCenter();
    private DiskDataCenter diskDataCenter = new DiskDataCenter();
    private Router router = new Router();
    private ConsumerManager consumerManager = new ConsumerManager(this);

    private final Object exchangeLocker = new Object();
    private final Object queueLocker = new Object();

    public VirtualHost(String virtualHostName) {
        this.virtualHostName = virtualHostName;
        diskDataCenter.init();
        try {
            memoryDataCenter.recovery(diskDataCenter);
        } catch (MqException | IOException e) {
            throw new RuntimeException(e);
        }
    }

    public MemoryDataCenter getMemoryDataCenter() {
        return memoryDataCenter;
    }

    public DiskDataCenter getDiskDataCenter() {
        return diskDataCenter;
    }

    public boolean exchangeDeclare(String exchangeName, ExchangeType exchangeType, boolean durable,
                                   boolean autoDelete, Map<String, Object> arguments) throws MqException {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange != null) {
                    throw new MqException("[VirtualHost] 当前交换机已存在！queueName=" + exchangeName);
                }
                exchange = new Exchange();
                exchange.setName(exchangeName);
                exchange.setType(exchangeType);
                exchange.setDurable(durable);
                exchange.setAutoDelete(autoDelete);
                exchange.setArguments(arguments);

                //先存硬盘，因为硬盘比较容易出现异常，如果是先存内存，硬盘后续产生异常，还需要删除内存中刚刚插入的数据
                if(durable) {
                    diskDataCenter.insertExchange(exchange);
                }
                memoryDataCenter.insertExchange(exchange);
                System.out.println("[VirtualHost] 交换机创建成功！queueName=" + exchangeName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 交换机创建失败！queueName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean exchangeDelete(String exchangeName) {
        exchangeName = virtualHostName + exchangeName;
        try {
            synchronized (exchangeLocker) {
                Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                if(exchange == null) {
                    throw new MqException("[VirtualHost] 待删除的交换机不存在！exchangeName=" + exchangeName);
                }
                //这里不用判断交换机是否存在，即使不存在，删除了也不会有什么副作用
                diskDataCenter.deleteExchange(exchangeName);
                memoryDataCenter.deleteExchange(exchangeName);
                System.out.println("[VirtualHost] 交换机删除成功！exchangeName=" + exchangeName);
            }
            return true;
        } catch(Exception e) {
            System.out.println("[VirtualHost] 交换机删除失败！exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDeclare(String queueName, boolean durable, boolean exclusive,
                                boolean autoDelete, Map<String, Object> arguments) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue != null) {
                    throw new MqException("[VirtualHost] 当前队列已存在！queueName=" + queueName);
                }
                queue = new MSGQueue();
                queue.setName(queueName);
                queue.setDurable(durable);
                queue.setExclusive(exclusive);
                queue.setAutoDelete(autoDelete);
                queue.setArguments(arguments);

                if(durable) {
                    diskDataCenter.insertQueue(queue);
                }
                memoryDataCenter.insertQueue(queue);
                System.out.println("[VirtualHost] 队列创建成功! queueName=" + queueName);
            }
            return true;
        } catch(Exception e) {
            System.out.println("[VirtualHost] 队列创建失败! queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueDelete(String queueName) {
        queueName = virtualHostName + queueName;
        try {
            synchronized (queueLocker) {
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null) {
                    throw new MqException("[VirtualHost] 当前队列不存在！queueName=" + queueName);
                }
                diskDataCenter.deleteQueue(queueName);
                memoryDataCenter.deleteQueue(queueName);
                System.out.println("[VirtualHost] 队列删除成功！queueName=" + queueName);
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 队列删除失败！queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueBind(String exchangeName, String queueName, String bindingKey) {
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(binding != null) {
                        throw new MqException("[VirtualHost] 当前绑定已存在！exchangeName=" + exchangeName +
                                ", queueName=" + queueName);
                    }
                    Exchange exchange = memoryDataCenter.getExchange(exchangeName);
                    if(exchange == null) {
                        throw new MqException("[VirtualHost] 绑定的交换机为空！exchangeName=" + exchangeName);
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(queueName);
                    if(queue == null) {
                        throw new MqException("[VirtualHost] 绑定的队列为空！queueName=" + queueName);
                    }
                    binding = new Binding();
                    binding.setExchangeName(exchangeName);
                    binding.setQueueName(queueName);
                    binding.setBindingKey(bindingKey);
                    if(exchange.isDurable() && queue.isDurable()) {
                        diskDataCenter.insertBinding(binding);
                    }
                    memoryDataCenter.insertBinding(binding);
                    System.out.println("[VirtualHost] 绑定创建成功！exchangeName=" + exchangeName +
                            ", queueName=" + queueName);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 绑定创建失败！exchangeName=" + exchangeName +
                    ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    public boolean queueUnBind(String exchangeName, String queueName) {
        exchangeName = virtualHostName + exchangeName;
        queueName = virtualHostName + queueName;
        try {
            synchronized (exchangeLocker) {
                synchronized (queueLocker) {
                    Binding binding = memoryDataCenter.getBinding(exchangeName, queueName);
                    if(binding == null) {
                        throw new MqException("[VirtualHost] 当前绑定不存在！exchangeName=" + exchangeName +
                                ", queueName=" + queueName);
                    }
                    diskDataCenter.deleteBinding(binding);
                    memoryDataCenter.deleteBinding(binding);
                    System.out.println("[VirtualHost] 绑定删除成功！exchangeName=" + exchangeName +
                            ", queueName=" + queueName);
                }
            }
            return true;
        } catch(Exception e) {
            System.out.println("[VirtualHost] 绑定删除失败！exchangeName=" + exchangeName +
                    ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 发布消息
     * 将消息发送给指定队列
     * @param exchangeName
     * @param basicProperties
     * @param body
     * @return
     */
    public boolean basicPublish(String exchangeName, String routingKey, BasicProperties basicProperties,
                                byte[] body) {
        //1.虚拟主机隔离
        exchangeName = virtualHostName + exchangeName;
        try {
            //2.获取交换
            Exchange exchange = memoryDataCenter.getExchange(exchangeName);
            if(exchange == null) {
                throw new MqException("[VirtualHost] 交换机不存在！exchangeName=" + exchangeName);
            }
            //3.校验 routingKey 是否合法
            if(!router.checkRoutingKey(routingKey)) {
                throw new MqException("[VirtualHost] routingKey 不合法！routingKey=" + basicProperties.getRoutingKey());
            }
            //4.根据交换机类型进行不同处理
            if(exchange.getType() == ExchangeType.Direct) {
                // 1) 直接交换机，这里约定队列名就是 routingKey(也是开发中最常见的用法)，将消息发送给指定的队列即可
                String queueName = virtualHostName + routingKey;
                Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                MSGQueue queue = memoryDataCenter.getQueue(queueName);
                if(queue == null) {
                    throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
                }
                sendMessage(queue, message);
            } else {
                // 2) 扇出交换机，将消息发送给每一个绑定的队列
                // 3) 主题交换机，在绑定的交换机中，发送给 bindingKey 和 routingKey 匹配的一组
                //首先都需要获取到该交换机的所有绑定
                ConcurrentHashMap<String, Binding> bindingNestMap = memoryDataCenter.getBindingMap(exchangeName);
                for(Map.Entry<String, Binding> entry : bindingNestMap.entrySet()) {
                    Binding binding = entry.getValue();
                    if(!router.checkBindingKey(binding.getBindingKey())) {
                        throw new MqException("[VirtualHost] bindingKey 不合法！bindingKey=" + binding.getBindingKey());
                    }
                    MSGQueue queue = memoryDataCenter.getQueue(binding.getQueueName());
                    if(queue == null) {
                        //这里就不抛异常类，不影响给其他队列发送消息
                        System.err.println("[VirtualHost] 队列不存在！queueName=" + binding.getQueueName());
                    }
                    Message message = Message.createMessageWithId(routingKey, basicProperties, body);
                    if(!router.route(routingKey, binding.getBindingKey())) {
                        continue;
                    }
                    sendMessage(queue, message);
                }
            }
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 发布消息失败！exchangeName=" + exchangeName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 发送消息到指定队列(前提是持久化)
     * @param queue
     * @param message
     */
    private void sendMessage(MSGQueue queue, Message message) throws IOException, InterruptedException {
        if(message.getDeliverMode() == 2) {
            diskDataCenter.sendMessage(queue, message);
        }
        memoryDataCenter.sendMessage(queue, message);
        System.out.println("[VirtualHost] 消息发布完成！queueName=" + queue.getName() +
                ", messageId=" + message.getMessageId());
        //通知当前队列的消费者可以进行消费了(唤醒阻塞线程)
        consumerManager.notifyConsumer(queue);
    }


    /**
     * 消费者订阅对应队列消息
     * @param consumerTag
     * @param queueName
     * @param autoAck
     * @param consumer
     * @return
     */
    public boolean basicConsume(String consumerTag, String queueName, boolean autoAck,
                                Consumer consumer) {
        queueName = virtualHostName + queueName;

        ConsumerEnv consumerEnv = new ConsumerEnv(consumerTag, queueName, autoAck, consumer);
        try {
            consumerManager.addConsumer(consumerEnv);
            System.out.println("[VirtualHost] 消息者订阅队列成功！consumerTag=" + consumerTag +
                    ", queueName=" + queueName);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息者订阅队列失败！consumerTag=" + consumerTag +
                    ", queueName=" + queueName);
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 主动应答
     * @param queueName
     * @return
     */
    public boolean basicAck(String queueName, String messageId) {
        queueName = virtualHostName + queueName;
        try {
            MSGQueue queue = memoryDataCenter.getQueue(queueName);
            if(queue == null) {
                throw new MqException("[VirtualHost] 队列不存在！queueName=" + queueName);
            }
            Message message = memoryDataCenter.getMessage(messageId);
            if(message == null) {
                throw new MqException("[VirtualHost] 消息不存在！messageId=" + messageId);
            }
            if(message.getDeliverMode() == 2) {
                diskDataCenter.deleteMessage(queue, message);
            }
            memoryDataCenter.removeMessage(messageId);
            memoryDataCenter.removeWaitAckMessage(queueName, messageId);
            System.out.println("[VirtualHost] 消息主动应答成功！queueName=" + queueName +
                    ", messageId=" + messageId);
            return true;
        } catch (Exception e) {
            System.out.println("[VirtualHost] 消息主动应答失败！queueName=" + queueName +
                    ", messageId=" + messageId);
            e.printStackTrace();
            return false;
        }
    }

}
